import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD

object test2 {
  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    conf.set("spark.testing.memory", "2147480000")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //设置顶点数据
    val vertexArray = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42)),
      (5L, ("Ed", 55)),
      (6L, ("Fran", 50))
    )
    val vertex: RDD[(VertexId, (String, Int))] = sc.parallelize(vertexArray)

    //设置边
    val edgeArray = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3)
    )
    val relationships: RDD[Edge[Int]] = sc.parallelize(edgeArray)

    //创建图
    val graph = Graph(vertex, relationships)

    //（2）使用mapEdges函数遍历所有的边，新增加一个属性值然后构建出新的图
    println("---------------------使用mapEdges函数遍历所有的边，新增加一个属性值然后构建出新的图------------------------")
    val graph2: Graph[(String,Int),String] = graph.mapEdges(e => (e.attr.toString + "ddd"))
    graph2.edges.collect.foreach(println(_))

    //（3）使用mapTriplets函数遍历所有的三元组，新增加一个属性值，然后返回新的图
    println("---------------------使用mapTriplets函数遍历所有的三元组，新增加一个属性值，然后返回新的图------------------------")
    var graph3: Graph[(String, Int), (String, Boolean)] = graph.mapTriplets(triplet => (triplet.attr.toString, true))
    graph3.edges.collect.foreach(println(_))



  }
}
